<?php

namespace App\Console\Commands\Canal;

use App\Kuafu\CouponExt\Model\OrderCouponExtModel;
use App\Kuafu\Order\Model\OrderCouponModel;
use App\Kuafu\Order\Model\OrderDetailModel;
use App\Kuafu\Order\Model\OrderDiscountModel;
use App\Kuafu\Order\Model\OrderModel;
use App\Kuafu\Shop\Model\ShopMonitorDataCHModel;
use App\Library\FeiShu\SendFs;
use Bavix\LaravelClickHouse\Database\Eloquent\Builder;
use Com\Alibaba\Otter\Canal\Protocol\EntryType;
use Com\Alibaba\Otter\Canal\Protocol\EventType;
use Com\Alibaba\Otter\Canal\Protocol\RowChange;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;
use xingwenge\canal_php\CanalClient;
use xingwenge\canal_php\CanalConnectorFactory;
use xingwenge\canal_php\Fmt;

class SyncPassengerToCh extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'canal:sync-passenger-to-clickhouse {--skip=0}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '同步客流数据至 clickhouse 
    --skip= : 跳过的 ACK 数量';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command
     */
    public function handle()
    {
        try {
            self::sendFs('The Canal has started.');

            $skip   = $this->option('skip');
            $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);

            $host = config('database.sync.canal.host');
            $client->connect($host);
            $client->checkValid();
            $client->subscribe(1, 'qa', 'kf_data_statistics.kf_shop_monitor_data'); # 修改过滤
            while (true) {
                $message = $client->getWithoutAck(100);

                if ($skip > 0) {
                    $client->ack($message->getId());
                    $skip--;
                    continue;
                }

                if ($entries = $message->getEntries()) {
                    foreach ($entries as $entry) {
                        self::setStartLog($entry->getHeader());
                        self::syncOrder($entry);
                        self::setEndLog($entry->getHeader());
//                        Fmt::println($entry);
                    }
                    $client->ack($message->getId());
                }

                sleep(1);
            }
        } catch (\Exception $e) {
            self::sendFs('订单同步报错了! 非捕捉的异常 [已经阻塞后续同步了!]', self::formatException($e));
        }

    }

    private static function syncOrder($entry)
    {
        switch ($entry->getEntryType()) {
            case EntryType::TRANSACTIONBEGIN:
            case EntryType::TRANSACTIONEND:
                return;
                break;
        }

        $rowChange = new RowChange();
        $rowChange->mergeFromString($entry->getStoreValue());
        $evenType = $rowChange->getEventType();
        $header   = $entry->getHeader();

        foreach ($rowChange->getRowDatas() as $rowData) {

            switch ($evenType) {
                // 新增
                case EventType::INSERT:
                    $insertData = self::fmtColumn($rowData->getAfterColumns());
                    ShopMonitorDataCHModel::query()->insert($insertData);
                    break;
                // 硬删除
                case EventType::DELETE:
//                    ShopMonitorDataCHModel::query($insertData);

                    break;
                // 更新
                case EventType::UPDATE:
                    break;
                default:
                    break;
            }
        }
    }

    private static function fmtColumn($columns)
    {
        $arr = [];
        foreach ($columns as $column) {
            $arr[$column->getName()] = $column->getValue();
        }
        return $arr;
    }

    private function sendFs($title, $content = '')
    {
        $this->info($title);
        Log::channel('orderSyncSingle')->error('同步数据库错误:', ['title' => $title, 'content' => $content]);
        SendFs::newRobot()->setTitleContent('[' . date('Y-m-d H:i:s') . '] : ' . $title, $content)->send();
    }

    private static function formatException(\Exception $e)
    {
        return json_encode([
            'line'    => $e->getLine(),
            'message' => $e->getMessage(),
            'trace'   => $e->getTrace()
        ], JSON_UNESCAPED_UNICODE);

    }

    private static function setStartLog($header)
    {
        Log::channel('orderSyncSingle')->info('同步数据库开始:', [
            sprintf("================> binlog[%s : %d],name[%s,%s], eventType: %s",
                $header->getLogfileName(),
                $header->getLogfileOffset(), $header->getSchemaName(), $header->getTableName(), $header->getEventType())
        ]);
    }

    private static function setEndLog($header)
    {
        Log::channel('orderSyncSingle')->info('同步数据库结束:', [
            sprintf("================> binlog[%s : %d],name[%s,%s], eventType: %s",
                $header->getLogfileName(),
                $header->getLogfileOffset(), $header->getSchemaName(), $header->getTableName(), $header->getEventType())
        ]);
    }
}
